package com.demo.java.zk;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.Stat;

import java.util.concurrent.TimeUnit;

public class ZkNodeTest {

    public static void main(String[] args) {
        CuratorFramework framework = CuratorFrameworkFactory.builder()
                .connectString("localhost:2181")
                .connectionTimeoutMs(15000)
                .sessionTimeoutMs(60000)
                .retryPolicy(new ExponentialBackoffRetry(5000, Integer.MAX_VALUE))
                .build();
        framework.start();

        String delimiter = "/";
        String rootNode = "/ocean";
        String masterNode = "master";
        String watcherNode = rootNode + delimiter + masterNode;

        String str1 = null;
        try {
            // TODO:// 第一个服务器启动， 在/ocean节点下创建master临时节点成功，data为服务器IP，则标识为master；
            str1 = framework.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.EPHEMERAL)
                    .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                    .forPath(watcherNode, "127.0.0.1".getBytes());
            System.out.println("node1 started, current is master node.");
        } catch (Exception e) {
            System.out.println("node1, master ephemeral exists.");
        }

        System.out.println("str1: " + str1);

        String str2 = null;
        try {
            // TODO:// 第二个服务器启动，在/ocean节点下创建master临时节点失败，则为非master；
            str2 = framework.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.EPHEMERAL)
                    .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                    .forPath(watcherNode, "127.0.0.2".getBytes());

        } catch (Exception e) {
            System.out.println("node2, master ephemeral exists.");
        }
        System.out.println("str2: " + str2);

        Stat stat = new Stat();
        byte[] data = new byte[0];
        try {
            // TODO:// 获取master节点数据, 查看master的IP地址
            data = framework.getData()
                    .storingStatIn(stat)
                    .forPath(watcherNode);
        } catch (Exception e) {
            System.out.println("get master data error.");
        }
        String dataStr = new String(data);
        System.out.println("master node data: " + dataStr);


        // /ocean/host1
        // /ocean/host2
        // TODO:// 监听ocean子节点的添加与删除, 监控服务器上线下线，
        PathChildrenCache pathChildrenCache = new PathChildrenCache(framework, rootNode, true);
        try {
            pathChildrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
            pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
                @Override
                public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
                    pathChildrenCacheEvent.getData();
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }

        // priorty>=1的优先调度, 调度不是RUNNING状态的WATING状态的任务。

        try {
            TimeUnit.SECONDS.sleep(100);
        } catch (InterruptedException e) {}

        framework.close();
    }
}
